/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package tuplestoreclient;

import core.ServerCallBack;
import core.AbstractCoreClient;
import core.Callback;
import core.FileCallBack;
import core.FileServerCallBack;
import core.ProtocolSD;
import core.ProtocolSD.IProtocolListener;
import core.Result;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.nio.channels.SelectionKey;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import logging.LOG;

/**
 *
 * @author Jorge
 */
public class TupleStoreClient extends AbstractCoreClient{

    private static final int MASTER_PORT=3000;

    private static  final String MASTER_ADDRESS="localhost";

    public TupleStoreClient() {
        LOG.append("A inicia cliente ....");
        addUnJoinParam(ProtocolSD.Operation.DATA);
    }


    @Override
    protected IProtocolListener createInstance() {
        return new ClienteListener();
    }

    public void getLocationToRead(final List<Integer>  list, Callback<Integer,Url> callback){
        connect(MASTER_ADDRESS, MASTER_PORT, new ServerCallBack<Integer,Url>(callback) {

            @Override
            public void onFinishClient(ProtocolSD protocolSD) {
                SelectionKey key = protocolSD.getKey();
                LOG.append("Pedido de servidores para GET "+list);
                protocolSD.write((byte)ProtocolSD.Operation.GET.ordinal(), list);
                key.interestOps(SelectionKey.OP_WRITE);
            }
        });
    }

    public void getLocationToWrite(final List<Integer>  list,final List<Integer> sizes, Callback<Integer,Url> callback){
        connect(MASTER_ADDRESS, MASTER_PORT, new ServerCallBack<Integer,Url>(callback) {

            @Override
            public void onFinishClient(ProtocolSD protocolSD) {
                SelectionKey key = protocolSD.getKey();
                LOG.append("Pedido de servidores para PUT"+list);
                protocolSD.write((byte)ProtocolSD.Operation.PUT_KEYS.ordinal(), list,sizes);
                key.interestOps(SelectionKey.OP_WRITE);
            }
        });
    }


    public void getData(final Map<Integer,Url> map,Callback<Integer,byte[]> callback){
        Map<String,List<Integer>> result = joinResult(map);
        for(Map.Entry<String,List<Integer>> entry : result.entrySet()){
            String [] host = entry.getKey().split("\\:");
            final List<Integer> keys = entry.getValue();
            connect(host[0], Integer.parseInt(host[1]), new ServerCallBack<Integer,byte[]>(callback) {

                @Override
                public void onFinishClient(ProtocolSD protocolSD) {
                    SelectionKey key = protocolSD.getKey();
                    LOG.append("Pedido de GET "+keys);
                    protocolSD.write((byte)ProtocolSD.Operation.GET.ordinal(), keys);
                    key.interestOps(SelectionKey.OP_WRITE);
                }
            });
        }
    }


    public void getData(final Map<Integer,Url> map,FileCallBack callback){
        Map<String,List<Integer>> result = joinResult(map);
        for(Map.Entry<String,List<Integer>> entry : result.entrySet()){
            String [] host = entry.getKey().split("\\:");
            final List<Integer> keys = entry.getValue();
            connect(host[0], Integer.parseInt(host[1]), new FileServerCallBack(callback) {
                
                @Override
                public void onFinishClient(ProtocolSD protocolSD) {
                    SelectionKey key = protocolSD.getKey();
                    LOG.append("Pedido de GET" +keys);
                    protocolSD.write((byte)ProtocolSD.Operation.GET.ordinal(), keys);
                    key.interestOps(SelectionKey.OP_WRITE);
                }
            });
        }
    }


    private static <K> Map<String,List<K>> joinResult(Map<K,Url> map){
        Map<String,List<K>> result= new HashMap<String, List<K>>();
        for(Map.Entry<K,Url> entry : map.entrySet()){
            String string = entry.getValue().getName();
            if(result.containsKey(string)){
                result.get(string).add(entry.getKey());
            }else{
                List<K>  list = new LinkedList<K>();
                list.add(entry.getKey());
                result.put(string, list);
            }
        }
        return result;
    }

    
    public void putData(Map<Integer,Url> map ,final Map<Integer,FileChannel> map1,Callback<Integer,Boolean> callback){
        Map<String,List<Integer>> result = joinResult(map);
        for(Map.Entry<String,List<Integer>> entry : result.entrySet()){
            String [] host = entry.getKey().split("\\:");
            final List<Integer> keys = entry.getValue();
            connect(host[0], Integer.parseInt(host[1]), new ServerCallBack<Integer,Boolean>(callback) {

                private Iterator<Map.Entry<Integer,FileChannel>> iterator;
                private Map.Entry<Integer,FileChannel> sendVal;
                private int operation;

                @Override
                public void onFinishClient(ProtocolSD protocolSD) {
                    SelectionKey key = protocolSD.getKey();
                    LOG.append("Pedido de PUT"+keys);
                    protocolSD.write((byte)ProtocolSD.Operation.PUT_KEYS.ordinal(),keys);
                    iterator = map1.entrySet().iterator();
                    key.interestOps(SelectionKey.OP_WRITE);
                }

                @Override
                public void onWriteComplete(ProtocolSD protocolSD) {
                    protocolSD.getKey().interestOps(SelectionKey.OP_READ);
                }

                @Override
                public void onReadAllArguments(ProtocolSD protocolSD) {
                    this.onReadAllArguments(protocolSD);
                }



                @Override
                public void onReadHeader(int operation, int numArgs, ProtocolSD protocolSD) {
                    this.operation=operation;
                    if(this.operation==ProtocolSD.Operation.DONE.ordinal()){
                        try {
                            //Mandar o primeiro ficheiro
                            protocolSD.getKey().interestOps(SelectionKey.OP_WRITE);
                            protocolSD.write(new byte[]{(byte) ProtocolSD.Operation.PUT_DATA.ordinal(), (byte) keys.size()});
                            sendVal = iterator.next();
                            int i = 4+ new Long(sendVal.getValue().size()).intValue();
                            protocolSD.writeInt(i);
                            protocolSD.writeInt(sendVal.getKey());
                            protocolSD.setFileChannel(sendVal.getValue(), new Long(sendVal.getValue().size()).intValue());
                        } catch (Exception ex) {
                            ex.printStackTrace();
                        }
                    }
                }
                
                @Override
                public void onReadArgument(byte[] data, ProtocolSD protocolSD){
                    this.onNewResult(ProtocolSD.byteArrayToInt(data, 0), data[4]==1);
                    if(iterator.hasNext() && operation==ProtocolSD.Operation.DONE.ordinal()){
                        try {
                            //Mandar o primeiro ficheiro
                            protocolSD.getKey().interestOps(SelectionKey.OP_WRITE);
                            sendVal = iterator.next();
                            protocolSD.writeInt(4 + new Long(sendVal.getValue().size()).intValue());
                            protocolSD.writeInt(sendVal.getKey());
                            protocolSD.setFileChannel(sendVal.getValue(), new Long(sendVal.getValue().size()).intValue());
                        } catch (IOException ex) {
                            ex.printStackTrace();
                        }
                    }                    
                }                
            });
        }

    }




    @Override
    public void handlerServerError(Throwable exception) {
        LOG.append("Erro no selector "+exception.getMessage());
    }

    @Override
    public void handlerFatalErrorCliente(Throwable throwable) {
        LOG.append("Erro na ligação com cliente "+throwable.getMessage());
    }

    @Override
    public void handleFinishConnect(SelectionKey key) {
        LOG.append("Nova ligação estabelecida");
    }




    public void put(final List<Integer> keys,final List<FileChannel> channels,final Callback<Integer,Boolean> callback){

        List<Integer> sizes = new ArrayList<Integer>(keys.size());
        for(FileChannel channel : channels){
            try {
                sizes.add(new Long(channel.size()).intValue());
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }

        this.getLocationToWrite(keys, sizes ,new Callback<Integer, Url>() {

            public void onError(Exception e) {
                callback.onError(e);
            }

            public void onResult(Result<Integer, Url> v) {
                System.out.println(v.getMap().keySet());
               if(v.getMap().size()!=0){
                Map<Integer,FileChannel> map1 = new HashMap<Integer, FileChannel>();
               for(Integer integer : v.getMap().keySet()){
                   int i = keys.indexOf(integer);

                   map1.put(integer, channels.get(i));
               }

               TupleStoreClient.this.putData(v.getMap(), map1,callback);
               }else{
                callback.onError(new Exception("Não foi retornado nenhum local para as chaves"));
               }
            }
        });
    }




    public void get(final List<Integer> keys,final FileCallBack callBack){

     this.getLocationToRead(keys, new Callback<Integer, Url>() {

            public void onError(Exception e) {
                callBack.onError(e);
            }

            public void onResult(Result<Integer, Url> v) {
               if(v.getMap().size()!=0){
               getData(v.getMap(), callBack);
               }else{
               callBack.onError(new Exception("Não foi possível obter localização das chaves"));
               }

            }
        });
    }
}
